package com.atguigu.userprofile.task;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constant.ConstCode;
import com.atguigu.userprofile.dao.TagInfoDao;
import com.atguigu.userprofile.util.MyPropsUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.connector.read.streaming.SparkDataStream;

import java.util.List;
import java.util.stream.Collectors;

/**
 * 将多个标签表合并为一张标签宽表
 *
 * 任务步骤:
 * 1. 获取外部传入的参数 taskId , busiDate
 *
 * 2. 明确有多少张标签表需要被合并 , 通过查询被计算的标签
 *
 * 3. 动态创建标签宽表
 *
 * 4. 组织SQL: insert(标签宽表) .... select(多个标签表)
 *
 * 5. 创建SparkSql环境， 执行SQL
 */
public class TaskMerge {

    public static void main(String[] args) {
        // 1. 获取外部传入的参数 taskId , busiDate
        String taskId = args[0] ;
        String busiDate = args[1] ;

        //2. 明确有多少张标签表需要被合并 , 通过查询被计算的标签
        List<TagInfo> tagInfos = TagInfoDao.selectTagInfosWithTaskStatusEnable();

        //3.动态创建标签宽表
        /*
            create table if not exists [upDbName].[tableName]
            (
               uid string ,
               [tagColumns]
            )
            row format delimited fields terminated by '\t'
            location '[hdfsPath]/[upDbName]/[tableName]'
         */
        String upDbName = MyPropsUtil.get(ConstCode.UP_DBNAME) ;
        String tableName = "up_merge_" + busiDate.replace("-", "") ;  // 2020-06-14 => 20200614
        String hdfsPath = MyPropsUtil.get(ConstCode.HDFS_STORE_PATH) ;

        //动态的处理宽表的列
        //标签宽表的标签列由实际计算的标签来决定
        // tagInfo =>  columnName columnType  =>  tag_code string  =>  tg_person_base_gender string
        String tagColumns = tagInfos.stream().map(
                tagInfo -> tagInfo.getTagCode().toLowerCase() + " string "
        ).collect(Collectors.joining(" , "));

        String createTable = " create table if not exists " + upDbName +  "." +  tableName +
                " (" +
                " uid string , " + tagColumns +
                " )" +
                " row format delimited fields terminated by '\\t'" +
                " location '" + hdfsPath+ "/" +  upDbName + "/" + tableName + "'" ;

        System.out.println("createTable : " + createTable);


        String dropTable = " drop table if exists " + upDbName + "." + tableName ;
        //4. 组织SQL: insert(标签宽表) .... select(多个标签表)
        /*
           insert overwrite table [upDbname].[tableName ]
           select * from
           (
             [
             select uid , 'tg_person_base_gender' as tag_code , tag_value from tg_person_base_gender where dt = '2020-06-14'
             union all
             select uid , 'tg_person_base_agegroup' as tag_code , tag_value from tg_person_base_agegroup where dt = '2020-06-14'
             ]
           )
           pivot ( max(tag_value) as tag_value for tag_code in ( ['tg_person_base_gender' , 'tg_person_base_agegroup' ] ))


           tagInfo => select uid ,'[tagCode]' as tag_code ,  tag_value from [tagCode] where dt = [busiDate]
                      select uid , 'tg_person_base_gender' as tag_code , tag_value from tg_person_base_gender where dt = '2020-06-14'
          */

        //动态的将每个表的查询语句处理好，并通过union all 合并
        String unionAllSql = tagInfos.stream().map(
                tagInfo -> "select uid, '" + tagInfo.getTagCode().toLowerCase() + "' as tag_code , tag_value from " + upDbName + "." + tagInfo.getTagCode().toLowerCase() + " where dt = '" + busiDate + "'"
        ).collect(Collectors.joining(" union all "));

        //动态将旋转值处理好
        String pivotValues = tagInfos.stream().map(
                tagInfo -> "'" + tagInfo.getTagCode().toLowerCase() + "'"
        ).collect(Collectors.joining(" , "));

        String insertSelectSql = " insert overwrite table " + upDbName + "." +  tableName +
                " select * from ( "+ unionAllSql +" )" +
                " pivot ( max(tag_value) as tag_value for tag_code in (" + pivotValues + "))" ;

        System.out.println("insertSelectSql : " + insertSelectSql);

        //5. 创建SparkSql环境， 执行SQL

        SparkConf sparkConf = new SparkConf().setAppName("task_merge_app");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //删除表: 考虑同一天跑多次的情况， 每次的标签不一样。
        sparkSession.sql(dropTable);
        //创建表
        sparkSession.sql(createTable);
        //写入数据: overwrite： 考虑幂等性。
        sparkSession.sql(insertSelectSql);
    }
}
